package icu.zlz.message.component;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import icu.zlz.common.core.constant.MqttConstants;
import icu.zlz.common.redis.service.RedisService;
import icu.zlz.msg.entity.bean.AlarmTopic;
import icu.zlz.msg.entity.bean.MqttMsg;
import icu.zlz.common.mongodb.entity.MqttRecord;
import icu.zlz.msg.entity.primary.*;
import icu.zlz.msg.repository.mongodb.MqttRecordRepository;
import icu.zlz.msg.repository.primary.IotMqttAlarmContactsRepository;
import icu.zlz.msg.repository.primary.IotMqttAlarmMsgRepository;
import icu.zlz.msg.repository.primary.IotTopicRepository;
import icu.zlz.msg.service.impl.AlarmService;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;

import static icu.zlz.common.core.constant.MqttConstants.MQTT_ALARM_TOPIC;

/**
 * 消息监听
 */
@Component
public class MQTTMsg {
    @Resource
    private MqttClient mqttClient;
    @Resource
    private AlarmService alarmService;
    @Resource
    private IotMqttAlarmMsgRepository iotMqttAlarmMsgRepository;
    @Resource
    private IotMqttAlarmContactsRepository iotMqttAlarmContactsRepository;
    @Resource
    private RedisService redisService;
    @Resource
    private MqttRecordRepository mqttRecordRepository;
    @Resource
    private IotTopicRepository iotTopicRepository;

    @PostConstruct
    public void init() {
        alarmService.initAlarm();
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {

            }

            /**
             * 处理mqtt信息
             * @param topic 主题
             * @param mqttMessage 信息
             * @throws Exception
             */
            @Override
            public void messageArrived(String topic, MqttMessage mqttMessage) throws UnsupportedEncodingException, JsonProcessingException {
                MqttMsg msg  =  toMqttMsg(mqttMessage);

                //保存记录
                saveMsgToMongoDB(msg);
                List<AlarmTopic> alarmTopics = redisService.getCacheList(MQTT_ALARM_TOPIC);
                for (AlarmTopic alarmTopic : alarmTopics) {
                    if (alarmTopic.getAlarmTopic().equals(msg.getTopic())) {
                        //发送告警消息
                        sendNotifications(alarmTopic, msg);
                        //保存告警记录
                        saveMsg(alarmTopic.getAlarmId(), msg);
                        break;
                    }
                }

            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

            }
        });
    }

    private MqttMsg toMqttMsg(MqttMessage mqttMessage) throws UnsupportedEncodingException, JsonProcessingException {
        String s=   new String(mqttMessage.getPayload(), MqttConstants.MSG_CODE);
        String payloadStr = s.split("\"payload\":\"")[1].split("\",")[0];
        String escapedPayloadStr = payloadStr.replace("\"", "\\\"")
                .replace("\t", "\\t")
                .replace("\n", "\\n");
        String newJsonString = s.replace(payloadStr, escapedPayloadStr);

        // 解析JSON字符串为Java对象
       return new ObjectMapper().readValue(newJsonString, MqttMsg.class);
    }

    private void saveMsgToMongoDB(MqttMsg msg) {
        int index = msg.getTopic().indexOf('/');
        if (index != -1) {
            // 如果找到了'/'，则提取'/'前后的部分
            String groupName = msg.getTopic().substring(0, index);
            String topicName = msg.getTopic().substring(index + 1);
            IotTopicEntity iotTopicEntity = iotTopicRepository.findByTopicNameAndGroupNameSingle(topicName, groupName);
            if (iotTopicEntity != null) {
                MqttRecord record = new MqttRecord();
                record.setPayload(msg.getPayload());
                record.setClientId(msg.getClientid());
                record.setTimestamp(LocalDateTime.ofInstant(Instant.ofEpochMilli(msg.getTimestamp()), ZoneId.systemDefault()));
                record.setTopicId(iotTopicEntity.getTopicId());
                record.setQos(msg.getQos());
                record.setUsername(msg.getUsername());
                record.setNode(msg.getNode());
                mqttRecordRepository.save(record);
            }
        }

    }

    private void saveMsg(Long mqttAlarmId, MqttMsg msg) {

        IotMqttAlarmMsgEntity save = new IotMqttAlarmMsgEntity();
        save.setAlarmId(mqttAlarmId);
        save.setMsg(msg.getPayload());
        save.setClientId(msg.getClientid());
        save.setTime(msg.getTimestamp());
        save.setQos(msg.getQos());
        save.setNode(msg.getNode());
        save.setUsername(msg.getUsername());
        iotMqttAlarmMsgRepository.save(save);
    }

    private void sendNotifications(AlarmTopic alarmTopic, MqttMsg mqttMsg)   {

        List<IotMqttAlarmContactsEntity> allByMqttAlarmId = iotMqttAlarmContactsRepository.findAllByAlarmIdAndStatus(alarmTopic.getAlarmId(), IotMqttAlarmContactsEntity.STATUS_NORMAL);
        for (IotMqttAlarmContactsEntity item : allByMqttAlarmId) {
            if (item.getNotificationMethod().equals("1") || item.getNotificationMethod().equals("3")) {
                //发送短信
                System.out.println("发送短信" + mqttMsg.getPayload());
            }
            if (item.getNotificationMethod().equals("2") || item.getNotificationMethod().equals("3")) {
                //发送邮件
                System.out.println("发送邮件" + mqttMsg.getPayload());
            }
        }

    }

}
